use async_trait::async_trait;
use fluviox_shared::{Result, FluvioXError, models::DataRecord};
use crate::traits::{SourceConnector, ConnectorConfig, ConnectorStatus, ConnectorMetrics};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use sqlx::{PgPool, Row, Column};
use std::collections::HashMap;
use uuid::Uuid;
use chrono::Utc;
use tracing::{info, error, warn};

/// 数据库数据源配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DatabaseSourceConfig {
    /// 数据库连接URL
    pub connection_url: String,
    /// 数据库类型
    pub database_type: DatabaseType,
    /// 查询SQL
    pub query: String,
    /// 增量查询配置
    pub incremental: Option<IncrementalConfig>,
    /// 轮询间隔（秒）
    pub poll_interval: u64,
    /// 批次大小
    pub batch_size: u32,
    /// 连接池配置
    pub pool_config: Option<PoolConfig>,
}

/// 数据库类型
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum DatabaseType {
    PostgreSQL,
    MySQL,
    SQLite,
    SQLServer,
}

/// 增量查询配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct IncrementalConfig {
    /// 增量字段名（通常是时间戳或ID）
    pub cursor_field: String,
    /// 初始游标值
    pub initial_cursor: Option<String>,
    /// 游标字段类型
    pub cursor_type: CursorType,
}

/// 游标类型
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum CursorType {
    Timestamp,
    Integer,
    String,
}

/// 连接池配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PoolConfig {
    /// 最大连接数
    pub max_connections: u32,
    /// 最小连接数
    pub min_connections: u32,
    /// 连接超时时间（秒）
    pub connect_timeout: u64,
}

impl ConnectorConfig for DatabaseSourceConfig {
    fn validate(&self) -> Result<()> {
        if self.connection_url.is_empty() {
            return Err(FluvioXError::Validation("数据库连接URL不能为空".to_string()));
        }
        
        if self.query.trim().is_empty() {
            return Err(FluvioXError::Validation("查询SQL不能为空".to_string()));
        }
        
        if self.poll_interval == 0 {
            return Err(FluvioXError::Validation("轮询间隔必须大于0".to_string()));
        }
        
        if self.batch_size == 0 {
            return Err(FluvioXError::Validation("批次大小必须大于0".to_string()));
        }
        
        Ok(())
    }
    
    fn to_json(&self) -> Result<Value> {
        Ok(serde_json::to_value(self)?)
    }
    
    fn from_json(value: Value) -> Result<Self> {
        Ok(serde_json::from_value(value)?)
    }
}

/// 数据库数据源连接器
pub struct DatabaseSourceConnector {
    config: Option<DatabaseSourceConfig>,
    pool: Option<PgPool>,
    status: ConnectorStatus,
    metrics: ConnectorMetrics,
    last_cursor: Option<String>,
}

impl DatabaseSourceConnector {
    pub fn new() -> Self {
        Self {
            config: None,
            pool: None,
            status: ConnectorStatus::Uninitialized,
            metrics: ConnectorMetrics::default(),
            last_cursor: None,
        }
    }
    
    /// 构建增量查询SQL
    fn build_incremental_query(&self, config: &DatabaseSourceConfig) -> String {
        if let Some(incremental) = &config.incremental {
            let cursor_value = self.last_cursor.as_ref()
                .or(incremental.initial_cursor.as_ref())
                .map(|v| v.as_str())
                .unwrap_or("0");
            
            let where_clause = match incremental.cursor_type {
                CursorType::Timestamp => {
                    format!("WHERE {} > '{}'", incremental.cursor_field, cursor_value)
                },
                CursorType::Integer => {
                    format!("WHERE {} > {}", incremental.cursor_field, cursor_value)
                },
                CursorType::String => {
                    format!("WHERE {} > '{}'", incremental.cursor_field, cursor_value)
                },
            };
            
            // 简单的SQL修改，在实际实现中应该使用SQL解析器
            if config.query.to_lowercase().contains("where") {
                format!("{} AND {}", config.query, where_clause.replace("WHERE", ""))
            } else {
                format!("{} {}", config.query, where_clause)
            }
        } else {
            config.query.clone()
        }
    }
    
    /// 从数据库行创建数据记录
    fn row_to_record(&self, row: &sqlx::postgres::PgRow) -> Result<DataRecord> {
        let mut data = serde_json::Map::new();
        
        // 遍历所有列
        for (i, column) in row.columns().iter().enumerate() {
            let column_name = column.name();
            
            // 根据列类型提取值（这里简化处理，实际应该根据具体类型）
            let value: Value = match row.try_get::<Option<String>, _>(i) {
                Ok(Some(v)) => Value::String(v),
                Ok(None) => Value::Null,
                Err(_) => {
                    // 尝试其他类型
                    match row.try_get::<Option<i64>, _>(i) {
                        Ok(Some(v)) => Value::Number(serde_json::Number::from(v)),
                        Ok(None) => Value::Null,
                        Err(_) => Value::String(format!("未知类型_{}", i)),
                    }
                }
            };
            
            data.insert(column_name.to_string(), value);
        }
        
        Ok(DataRecord {
            id: Uuid::new_v4(),
            topic: "database-source".to_string(),
            partition: 0,
            offset: 0,
            timestamp: Utc::now(),
            data: Value::Object(data),
            metadata: HashMap::new(),
        })
    }
}

#[async_trait]
impl SourceConnector for DatabaseSourceConnector {
    type Config = DatabaseSourceConfig;
    
    fn name(&self) -> &str {
        "database-source"
    }
    
    fn version(&self) -> &str {
        "1.0.0"
    }
    
    fn description(&self) -> &str {
        "数据库数据源连接器，支持PostgreSQL、MySQL等，支持增量同步"
    }
    
    async fn initialize(&mut self, config: Self::Config) -> Result<()> {
        info!("初始化数据库数据源连接器");
        
        // 验证配置
        config.validate()?;
        
        // 创建连接池（目前只支持PostgreSQL）
        match config.database_type {
            DatabaseType::PostgreSQL => {
                let pool = PgPool::connect(&config.connection_url).await
                    .map_err(|e| FluvioXError::Database(e))?;
                self.pool = Some(pool);
            },
            _ => {
                return Err(FluvioXError::Internal("暂不支持的数据库类型".to_string()));
            }
        }
        
        // 设置初始游标
        if let Some(incremental) = &config.incremental {
            self.last_cursor = incremental.initial_cursor.clone();
        }
        
        self.config = Some(config);
        self.status = ConnectorStatus::Initialized;
        
        info!("数据库数据源连接器初始化完成");
        Ok(())
    }
    
    async fn test_connection(&self) -> Result<bool> {
        info!("测试数据库连接");
        
        if let Some(pool) = &self.pool {
            match sqlx::query("SELECT 1").fetch_one(pool).await {
                Ok(_) => {
                    info!("数据库连接测试成功");
                    Ok(true)
                },
                Err(e) => {
                    error!("数据库连接测试失败: {}", e);
                    Ok(false)
                }
            }
        } else {
            warn!("数据库连接池未初始化");
            Ok(false)
        }
    }
    
    async fn start(&mut self) -> Result<()> {
        info!("启动数据库数据源连接器");
        self.status = ConnectorStatus::Running;
        Ok(())
    }
    
    async fn stop(&mut self) -> Result<()> {
        info!("停止数据库数据源连接器");
        self.status = ConnectorStatus::Stopped;
        Ok(())
    }
    
    async fn read_records(&mut self, batch_size: usize) -> Result<Vec<DataRecord>> {
        let config = self.config.as_ref()
            .ok_or_else(|| FluvioXError::Internal("连接器未初始化".to_string()))?;
        
        let pool = self.pool.as_ref()
            .ok_or_else(|| FluvioXError::Internal("数据库连接池未初始化".to_string()))?;
        
        // 构建查询SQL
        let query_sql = self.build_incremental_query(config);
        let limited_query = format!("{} LIMIT {}", query_sql, batch_size);
        
        info!("执行查询: {}", limited_query);
        
        match sqlx::query(&limited_query).fetch_all(pool).await {
            Ok(rows) => {
                let mut records = Vec::new();
                let mut max_cursor: Option<String> = None;
                
                for row in rows {
                    let record = self.row_to_record(&row)?;
                    
                    // 更新游标值
                    if let Some(incremental) = &config.incremental {
                        if let Some(cursor_value) = record.data.get(&incremental.cursor_field) {
                            let cursor_str = match cursor_value {
                                Value::String(s) => s.clone(),
                                Value::Number(n) => n.to_string(),
                                _ => cursor_value.to_string(),
                            };
                            max_cursor = Some(cursor_str);
                        }
                    }
                    
                    records.push(record);
                }
                
                // 更新最后的游标值
                if let Some(cursor) = max_cursor {
                    self.last_cursor = Some(cursor);
                }
                
                // 更新指标
                self.metrics.records_processed += records.len() as u64;
                self.metrics.last_activity = Some(Utc::now());
                
                info!("成功读取 {} 条记录", records.len());
                Ok(records)
            },
            Err(e) => {
                self.metrics.errors_count += 1;
                error!("数据库查询失败: {}", e);
                Err(FluvioXError::Database(e))
            }
        }
    }
    
    fn status(&self) -> ConnectorStatus {
        self.status.clone()
    }
    
    fn metrics(&self) -> ConnectorMetrics {
        self.metrics.clone()
    }
}
